package beautiful.butterfly.drds.data_exchange.core.statistics.container.communicator.job;


import beautiful.butterfly.drds.data_exchange.core.statistics.communication.Communication;
import beautiful.butterfly.drds.data_exchange.core.statistics.communication.CommunicationTool;
import beautiful.butterfly.drds.data_exchange.core.statistics.container.collector.ProcessInnerCollector;
import beautiful.butterfly.drds.data_exchange.core.statistics.container.communicator.AbstractContainerCommunicator;
import beautiful.butterfly.drds.data_exchange.core.statistics.container.report.ProcessInnerReporter;
import beautiful.butterfly.drds.data_exchange.core.util.container.Constants;
import beautiful.butterfly.drds.data_exchange.dataxservice.face.domain.enums.State;
import common.util.Configuration;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;

@Slf4j
public class StandAloneJobContainerCommunicator extends AbstractContainerCommunicator
{


    public StandAloneJobContainerCommunicator(Configuration configuration)
    {
        super(configuration);
        super.setCollector(new ProcessInnerCollector(configuration.getLong(
                Constants.datax_core_container_job_id)));
        super.setReporter(new ProcessInnerReporter());
    }

    @Override
    public void registerCommunication(List<Configuration> configurationList)
    {
        super.getCollector().registerTGCommunication(configurationList);
    }

    @Override
    public Communication collect()
    {
        return super.getCollector().collectFromTaskGroup();
    }

    @Override
    public State collectState()
    {
        return this.collect().getState();
    }

    /**
     * 和 DistributeJobContainerCollector 的 report 实现一样
     */
    @Override
    public void report(Communication communication)
    {
        super.getReporter().reportJobCommunication(super.getJobId(), communication);

        log.info(CommunicationTool.Stringify.getSnapshot(communication));
        reportVmInfo();
    }

    @Override
    public Communication getCommunication(Integer taskGroupId)
    {
        return super.getCollector().getTGCommunication(taskGroupId);
    }

    @Override
    public Map<Integer, Communication> getCommunicationMap()
    {
        return super.getCollector().getTGCommunicationMap();
    }
}
